package com.jadows.tdog.message.handler.impl;

import cn.hutool.core.collection.CollUtil;
import com.jadows.tdog.constants.CacheConstants;
import com.jadows.tdog.constants.GlobalActions;
import com.jadows.tdog.domain.ChatMessage;
import com.jadows.tdog.message.WsReqWrapper;
import com.jadows.tdog.message.details.AbstractMessage;
import com.jadows.tdog.message.details.event.MessageReadEvent;
import com.jadows.tdog.message.handler.IMqMessageHandler;
import com.jadows.tdog.server.codec.JsonWebSocketFrame;
import com.jadows.tdog.state.OnlineUserRegister;
import com.jadows.tdog.state.SimpleWSSessionUtil;
import io.netty.channel.Channel;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;

import java.util.Set;

/**
 * 消息已读处理器
 *
 * @author taojiacheng
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class MessageReadHandler implements IMqMessageHandler<ChatMessage> {

    final MongoTemplate mongoTemplate;
    final OnlineUserRegister onlineUserRegister;
    final RedissonClient redissonClient;

    /**
     * 订阅已读消息
     */
    @PostConstruct
    protected void subscribeReadMessage() {
        String instanceId = onlineUserRegister.getCurrentInstanceId();
        RTopic rTopic = redissonClient.getTopic(CacheConstants.DIRECT_TEXT_MSG_READ_TOPIC + instanceId);
        rTopic.addListener(ChatMessage.class, (channel, msg) -> {
            this.arrived(msg);
        });
    }


    public void readMessage(WsReqWrapper<AbstractMessage> reqWrapper) {
        AbstractMessage message = reqWrapper.getData();
        Long messageId = message.getMessageId();
        // 推送已读消息
        ChatMessage chatMessage = mongoTemplate.findOne(Query.query(Criteria.where("messageId")
                .is(messageId)), ChatMessage.class);
        if (chatMessage == null) {
            log.warn("消息已读, 消息不存在: {}", messageId);
            return;
        }
        Update update = Update.update("isRead", true);
        mongoTemplate.updateMulti(Query.query(
                Criteria.where("messageId")
                        .lte(messageId)
                        .and("unionId")
                        .is(chatMessage.getUnionId())
        ), update, ChatMessage.class);
        this.publish(chatMessage);
    }


    @Override
    public Long publish(ChatMessage chatMessage) {
        // 通知消息发起方，消息已读
        String fromUid = chatMessage.getFromUid();
        Set<String> userInstanceIds = onlineUserRegister.getUserInstanceId(fromUid);
        // 判断目标用户是否在线
        if (CollUtil.isEmpty(userInstanceIds)) {
            // 不在线 不再推送
            log.info("通知消息已读, 目标用户 {} 不在线.", fromUid);
            return -1L;
        }
        for (String instanceId : userInstanceIds) {
            // 推送已读消息给目标用户
            String topicName = CacheConstants.DIRECT_TEXT_MSG_READ_TOPIC + instanceId;
            RTopic topic = redissonClient.getTopic(topicName);
            topic.publish(chatMessage);
        }
        return chatMessage.getMessageId();
    }

    @Override
    public void arrived(ChatMessage message) {
        // 消息投递到这里，就是当前用户的ws连接就在这台服务器上
        log.info("已读消息处理: {}", message);
        String fromUid = message.getFromUid();
        Set<Channel> channels = SimpleWSSessionUtil.getChannelByUid(fromUid);
        if (CollUtil.isEmpty(channels)) {
            log.warn("已读消息处理: 目标用户 {} 不在线.", fromUid);
            return;
        }
        // 当前用户在线，推送已读消息
        WsReqWrapper<AbstractMessage> reqWrapper = new WsReqWrapper<>();
        reqWrapper.setAction(GlobalActions.DIRECT_TEXT_MSG_READ);
        AbstractMessage messageReadEvent = new MessageReadEvent().setToUid(message.getToUid())
                .setMessageId(message.getMessageId())
                .setFromUid(message.getFromUid())
                .setSendTime(message.getSendTime());
        reqWrapper.setData(messageReadEvent);
        reqWrapper.setSeq(System.currentTimeMillis() + "");
        channels.forEach(wsChannel ->
                wsChannel.writeAndFlush(new JsonWebSocketFrame(reqWrapper)));
    }
}
